/*
 * Licensed to the 南京汇海交通科技有限公司
 *
 * Copyright (c) 2025.
 * All rights reserved.
 *
 * 本软件属南京汇海交通科技有限公司所有，在未获得南京汇海交通科技有限公司正式授权情况下，任何企业和个人，不能获取、阅读、安装、传播本软件涉及的任何受知识产权保护的内容。
 */
package com.chframework.common.thread;

//import com.google.common.collect.Lists;

import com.chframework.common.thread.threadpool.ThreadPoolHelper;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.ForkJoinTask;
import java.util.concurrent.RecursiveTask;
//import java.util.stream.Collectors;

/**
 * 分治，顾名思义，即分而治之，是一种解决复杂问题的思维方法和模式；
 * 具体来讲，指的是 把一个复杂的问题分解成多个相似的子问题，然后再把子问题分解成更小的子问题，直到子问题简单到可以直接求解。
 * 理论上来讲，解决每一个问题都对应着一个任务，所以对于问题的分治，实际上就是对于任务的分治。
 * <p>
 * 分治思想在很多领域都有广泛的应用，例如：
 * 算法领域有分治算法（归并排序、快速排序都属于分治算法，二分法查找也是一种分治算法）；
 * 大数据领域知名的计算框架MapReduce背后的思想也是分治。
 * <p>
 * 分治任务模型可分为两个阶段：
 * 一个阶段是 【任务分解】，也就是将任务迭代地分解为子任务，直至子任务可以直接计算出结果；
 * 另一个阶段是 【结果合并】，即逐层合并子任务的执行结果，直至获得最终结果。
 * <p>
 * 在分治任务模型里，任务和分解后的子任务具有相似性，这种相似性往往体现在任务和子任务的算法是相同的，
 * 但是计算的数据规模是不同的。具备这种相似性的问题，我们往往都采用递归算法。
 * <p>
 * 在Java并发包里提供了一种叫做Fork/Join的并行计算框架，主要就是用来支持分治任务模型的。
 * 这个计算框架里的 Fork 对应的是分治任务模型里的任务分解，Join 对应的是结果合并。
 * <p>
 * Fork/Join计算框架主要包含两部分：
 * 一部分是 分治任务的线程池ForkJoinPool；
 * 另一部分是 分治任务ForkJoinTask。
 * <p>
 * ForkJoinTask是一个抽象类，它的方法有很多，最核心的是fork()方法和join()方法，
 * 其中fork()方法会异步地执行一个子任务，
 * 而join()方法则会阻塞当前线程来等待子任务的执行结果。
 * <p>
 * ForkJoinTask有两个子类，即 RecursiveAction 和 RecursiveTask，它们都是用递归的方式来处理分治任务的。
 * 这两个子类都定义了抽象方法compute()，不过区别是
 * RecursiveAction定义的compute()没有返回值，
 * 而RecursiveTask定义的compute()方法是有返回值的。
 * 这两个子类也是抽象类，在使用的时候，需要我们通过定义子类的方式进行扩展。
 * <p>
 * Fork/Join并行计算框架的核心组件是ForkJoinPool。
 * ForkJoinPool支持任务窃取机制，能够让所有线程的工作量基本均衡，
 * 不会出现有的线程很忙，而有的线程很闲的状况，所以性能很好。
 * Java 1.8提供的Stream API里面并行流parallelStream()也是以ForkJoinPool为基础的。
 * <p>
 * 需要注意的是，
 * 默认情况下所有的并行流计算都共享一个ForkJoinPool，这个共享的ForkJoinPool默认的线程数是CPU的核数；
 * 如果所有的并行流计算都是CPU密集型计算的话，完全没有问题，
 * 但是如果存在I/O密集型的并行流计算，那么很可能会因为一个很慢的I/O计算而拖慢整个系统的性能。
 * 所以，建议用不同的ForkJoinPool执行不同类型的计算任务！
 * <p>
 * 拓展 - ParallelStream最佳实践
 * 针对IO密集型的操作，应尽量使用ThreadPool，而不是ForkJoin、parallelStream。
 * 避免在没有自定义ForkJoinPool的情况下使用Collection.parallelStream()或Stream.parallel()。
 * 因为共用CommonPool可能导致多个独立的业务操作之间相互争抢线程以至相互影响。
 * 优先避免使用，如果一定要用，请指定特定的ForkJoinPool。
 * <p>
 * JDK内置的CommonPool内线程初始化时机可能非常早，相应线程上绑定的ClassLoader可能并非应用真正使用的ClassLoader，
 * 这对于依赖特定ClassLoader加载类的框架来说，可能出现意外的找不到类问题。
 * <p>
 * 使用示例：
 * List<Integer> list = Lists.newArrayList(1, 2, 3, 4, 5, 6, 7, 8, 9);
 * ForkJoinPool forkJoinPool = ThreadPoolHelper.buildForkJoinPool();
 * ForkJoinTask<List<Integer>> task = forkJoinPool.submit(() -> list.parallelStream()
 *     .map(x -> {
 *         log.debug(Thread.currentThread().getName() + " ： " + forkJoinPool.getActiveThreadCount() + " - " + forkJoinPool.getQueuedTaskCount());
 *         return x + 1;
 *     })
 *     .collect(Collectors.toList())
 * );
 * log.debug((task.join());
 * forkJoinPool.shutdown();
 *
 * 针对多线程的异步场景，
 * 对ForkJoinPool理解不深入的情况下，请谨慎使用！
 * 使用它批量处理比较耗时的大任务时，不合理的编码组织很可能导致无法达到预期的效果！
 */
public class ForkJoinHelper {

    /**
     * 使用自定义 ForkJoinPool 执行分治任务
     */
    public static <T> T execute(ForkJoinTask<T> task) {
        ForkJoinPool forkJoinPool = ThreadPoolHelper.buildForkJoinPool();
        T result = forkJoinPool.invoke(task);
        if (!forkJoinPool.isShutdown()) {
            forkJoinPool.shutdown();
        }
        return result;
    }

    /**
     * 使用自定义 ForkJoinPool 执行分治任务，有返回值
     */
    public static <V> V execute(Callable<V> callable) {
        ForkJoinPool forkJoinPool = ThreadPoolHelper.buildForkJoinPool();
        ForkJoinTask<V> task = forkJoinPool.submit(callable);
        V result = task.join();
        if (!forkJoinPool.isShutdown()) {
            forkJoinPool.shutdown();
        }
        return result;
    }

    /**
     * 使用自定义 ForkJoinPool 执行分治任务，无返回值
     */
    public static void execute(Runnable runnable) {
        ForkJoinPool forkJoinPool = ThreadPoolHelper.buildForkJoinPool();
        ForkJoinTask<?> task = forkJoinPool.submit(runnable);
        task.join();
        if (!forkJoinPool.isShutdown()) {
            forkJoinPool.shutdown();
        }
    }

//CHECKSTYLE:OFF

    /**
     * 示例1 - 递归任务：计算斐波那契数列
     */
    static class Fibonacci extends RecursiveTask<Integer> {
        final int n;

        Fibonacci(int n) {
            this.n = n;
        }

        protected Integer compute() {
            if (n <= 1) {
                return n;
            }
            Fibonacci f1 = new Fibonacci(n - 1);
            //创建子任务
            f1.fork();
            Fibonacci f2 = new Fibonacci(n - 2);
            //等待子任务结果，并合并结果
            return f2.compute() + f1.join();
        }
    }

    /**
     * 示例2 - 递归任务：模拟MapReduce统计单词数量
     */
    static class MR extends RecursiveTask<Map<String, Long>> {
        private String[] fc;
        private int start, end;

        //构造函数
        MR(String[] fc, int fr, int to) {
            this.fc = fc;
            this.start = fr;
            this.end = to;
        }

        @Override
        protected Map<String, Long> compute() {
            if (end - start == 1) {
                return calc(fc[start]);
            } else {
                int mid = (start + end) / 2;
                MR mr1 = new MR(fc, start, mid);
                mr1.fork();
                MR mr2 = new MR(fc, mid, end);
                //计算子任务，并返回合并的结果
                return merge(mr2.compute(), mr1.join());
            }
        }

        /**
         * 合并结果
         */
        private Map<String, Long> merge(Map<String, Long> r1, Map<String, Long> r2) {
            Map<String, Long> result = new HashMap<>(r1);
            //合并结果
            r2.forEach((k, v) -> result.merge(k, v, Long::sum));
            return result;
        }

        /**
         * 统计单词数量
         */
        private Map<String, Long> calc(String line) {
            Map<String, Long> result = new HashMap<>();
            //分割单词
            String[] words = line.split("\\s+");
            //统计单词数量
            for (String w : words) {
                Long v = result.get(w);
                if (v != null) {
                    result.put(w, v + 1);
                } else {
                    result.put(w, 1L);
                }
            }
            return result;
        }
    }

    /*public static void main(String[] args) {
        // 示例1 - 测试
        //创建分治任务
        Fibonacci fib = new Fibonacci(30);
        //创建分治任务线程池
        Integer result1 = ForkJoinHelper.execute(fib);
        //输出结果
        System.out.println(result1);

        // 示例2 - 测试
        String[] fc = {"hello world", "hello me", "hello fork", "hello join", "fork join in world"};
        //创建任务
        MR mr = new MR(fc, 0, fc.length);
        //执行任务
        Map<String, Long> result2 = ForkJoinHelper.execute(mr);
        //输出结果
        result2.forEach((k, v) -> System.out.println(k + ":" + v));

        // 示例3 - 测试
        List<Integer> list = Lists.newArrayList(1, 2, 3, 4, 5, 6, 7, 8, 9);
        ForkJoinPool forkJoinPool = ThreadPoolHelper.buildForkJoinPool();
        ForkJoinTask<List<Integer>> task = forkJoinPool.submit(() -> list.parallelStream()
                .map(x -> {
                    System.out.println(Thread.currentThread().getName() + " ： " + forkJoinPool.getActiveThreadCount() + " - " + forkJoinPool.getQueuedTaskCount());
                    return x + 1;
                })
                .collect(Collectors.toList())
        );
        System.out.println(task.join());
        forkJoinPool.shutdown();
    }*/

//CHECKSTYLE:ON
}
